package com.chukun.multi.pipeline.reusable;

import com.chukun.multi.twophase.AbstractTerminatableThread;
import com.chukun.multi.twophase.TerminationToken;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 基于工作者线程的Pipe实现类
 * 提交到该Pipe的任务由指定个数的工作线程共同处理
 * @param <IN>
 * @param <OUT>
 */
public class WorkThreadPipeDecorator<IN,OUT> implements Pipe<IN,OUT> {

    private final BlockingQueue<IN> workQueue;
    private final Set<AbstractTerminatableThread> workThreads = new HashSet<>();

    private final Pipe<IN,OUT> delegate;

    private final TerminationToken terminationToken = new TerminationToken();

    public WorkThreadPipeDecorator(BlockingQueue<IN> workQueue,Pipe<IN,OUT> delegate,int workCount) {
        if (workCount<0) {
            throw new IllegalArgumentException("workCount should be positive");
        }
        this.workQueue = workQueue;
        this.delegate = delegate;
        for (int i = 0; i < workCount; i++) {
            workThreads.add(new AbstractTerminatableThread(terminationToken) {
                @Override
                protected void doRun() throws Exception {
                    try{
                        dispatch();
                    }finally {
                        terminationToken.reservations.incrementAndGet();
                    }
                }
            });
        }
    }

    protected void dispatch() throws InterruptedException {
        IN input = workQueue.take();
        delegate.process(input);
    }

    @Override
    public void setNextPipe(Pipe<IN, OUT> nextPipe) {
        delegate.setNextPipe(nextPipe);
    }

    @Override
    public void init(PipeContext pipeCtx) {
        delegate.init(pipeCtx);
        for (AbstractTerminatableThread workThread : workThreads) {
            workThread.start();
        }
    }

    @Override
    public void shutdown(long timeout, TimeUnit unit) {
        for (AbstractTerminatableThread workThread : workThreads) {
            workThread.terminate();
            try {
                workThread.join(TimeUnit.MICROSECONDS.convert(timeout,unit));
            } catch (InterruptedException e) {
            }
        }
        delegate.shutdown(timeout,unit);
    }

    @Override
    public void process(IN input) throws InterruptedException {
        workQueue.put(input);
        terminationToken.reservations.incrementAndGet();
    }
}
